package com.myflink.day05;

import com.myflink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;

/**
 * @author Shelly An
 * @create 2020/9/22 15:54
 * 用户自定义watermark：
 * 周期性AssignerWithPeriodicWatermarks
 */
public class Watermark_Periodic {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //指定时间语义
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //设置周期性生成watermark的时间间隔，默认200ms，一般不会改动
        env.getConfig().setAutoWatermarkInterval(5000L);

        env
                .socketTextStream("localhost", 4444)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
                    }
                })
                //从时间流里抽取事件时间 官方提供的
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<WaterSensor>() {
                    //保证watermark单调不减
                    private Long maxTs = Long.MIN_VALUE;

                    /**
                     * 固定时间间隔就生成一次，不断的生成，与是否有数据无关
                     * @return watermark
                     */
                    @Nullable
                    @Override
                    public Watermark getCurrentWatermark() {
                        System.out.println("周期性...");
                        return new Watermark(maxTs);
                    }

                    /**
                     * 抽取时间戳
                     * @param element 事件
                     * @param previousElementTimestamp 上一个事件的时间戳
                     * @return 时间戳
                     */
                    @Override
                    public long extractTimestamp(WaterSensor element, long previousElementTimestamp) {
                        maxTs = Math.max(maxTs, element.getTs() * 1000L);
                        return element.getTs() * 1000L;
                    }
                })
                .keyBy(WaterSensor::getId)
                .timeWindow(Time.seconds(5))
                .process(
                        new ProcessWindowFunction<WaterSensor, Long, String, TimeWindow>() {
                            /**
                             * 计数
                             * @param s key
                             * @param context 上下文
                             * @param elements 整个窗口本组（key：s）的数据
                             * @param out 采集器
                             * @throws Exception
                             */
                            @Override
                            public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<Long> out) throws Exception {
                                for (WaterSensor element : elements) {
                                    System.out.println(element.toString());
                                }
                                out.collect(elements.spliterator().estimateSize());
                            }
                        }).print();
        env.execute();
    }
}
